// Copyright (C) 2010 and onwards Noam Shazeer and Georges Harik
// Pieces of this code may be licenced under the Apache License, version 2.0 
//   from Harik Shazeer Labs

#include <fstream>
#include <unistd.h>
#include <signal.h>
#include "jobqueue.h"
#include "economy.h"
#include "tokenizer.h"
#include "lexicon.h"
#include "dataset.h"
#include "corpus.h"
#include "timer.h"

Economy * G_EC;

uint64 kMinWeightPerWord = 0; // read later from config

void * EconomyMaintenance(void * junk) {
  G_EC->RunScheduledCommands();
  return NULL;
}

void InitEconomy() {
  // cout << "::InitEconomy" << endl;
  if (G_EC) return;
  InitInterpret();
  InitMath();
  InitObjects();
  InitFunctionModel();
  kMinWeightPerWord = (uint64)ConfigDouble("min_weight_per_word");
  G_EC = new Economy;
  G_EC->Construct();
  G_SYNC->AddMaintenanceRoutine(&EconomyMaintenance);

}

void CreditAssigner::Initialize(double time, double utility) {
  history_.clear();
  history_.push_back(make_pair(time, utility));
  horizon_ = 0;
  horizon_fraction_ = 0.3;
}

void CreditAssigner::PushHorizon() {
  while (ShouldPushHorizon()) horizon_++; }

bool CreditAssigner::ShouldPushHorizon() {
  if (horizon_ + 1 >= history_.size()) return false;
  int64 new_h = horizon_ + 1;
  double t = history_[new_h].first;
  double begin_t = history_[0].first;
  double end_t = history_.back().first;
  if (begin_t == end_t) return false;
  double fraction_from_end = (end_t - t) / (end_t - begin_t);
  return (fraction_from_end > horizon_fraction_);
}

double CreditAssigner::Credit(double time, double utility) {  
  if (history_.size() == 0) {
    Initialize(time, utility);
    return 0;
  }
  CHECK(history_.size() > 0);
  CHECK(time >= history_.back().first);
  CHECK(utility > history_.back().second);
  double improvement = utility - history_.back().second;
  history_.push_back(make_pair(time, utility));
  PushHorizon();
  if (time == history_[0].first) return 0;
  double slope = 
    (utility - history_[horizon_].second) 
    / (time - history_[horizon_].first);
  CHECK(finite(slope));
  double ret = improvement / slope;
  /*
  cout << "Credit " 
       << " time = " << time << endl
       << " utility = " << utility << endl
       << " improvement = " << improvement << endl
       << " horizon = " << history_[horizon_].first << " , " 
       << history_[horizon_].second << endl
       << " slope = " << slope << endl
       << " credit = " << ret << endl;
  */
  return ret;
}

Economy::Economy() {
  pthread_mutex_init(&lock_, NULL);
  pthread_mutex_init(&credit_lock_, NULL);
  pthread_mutex_init(&commands_lock_, NULL);
  accept_threshold_ = 0;
  G_ALLOC->AddDependant(this);
}

void Economy::Construct() {

  // Load the libaries into the global space
  {
    Vinterpreter vi;
    ThreadAllocator a;
    vi.a_= &a;
    vi.LoadFile("library", false, kEconomyMode);
  }
  
  Problem *problem = NULL;

  if (G_CONFIG->Exists("explicit_problem_files")) {
    problem = new ExplicitProblem(ConfigString("explicit_problem_files"));;
  } else if (G_CONFIG->Exists("int_sequence_file")) {
    problem = new IntSequenceProblem;
  } else if (G_CONFIG->Exists("corpus_file")) {
    problem = new SequenceProblem;
  }
  else {
    cerr << "Error - no problem specified";
    CHECK(false);
  }

  string prefix = "";
  bool start_from = G_CONFIG->Exists("start_from");
  if (start_from) prefix = G_CONFIG->GetString("start_from");
  solution_ =
    start_from ? new Solution(problem, prefix) : new Solution(problem);

  // Read the initial entities
  {
    EconomyThread et;
    ThreadAllocator a;
    et.SetThreadAllocator(&a);
    et.ReadInitialAds("entities");
  }
  if (start_from) {
    ad_store_.Load(prefix + ".ads");
    uint64 time;
    File::Read(prefix + ".time", Memblock((const char *)&time, sizeof(uint64)));
    clock_.Set(time);
  }
  utility_ = solution_->Utility();
  assigner_.Initialize(clock_.Get(), utility_);
  cout << "Initial utility: " << Delimit(utility_) << endl;
}

void Economy::Store(string prefix) {
  cout << "::Economy::Store" << endl;
  uint64 time0 = TimeNS();

  // Store the memory, ads, and model
  G_ALLOC->Store(prefix + ".memory");
  ad_store_.Store(prefix + ".ads");
  solution_->Store(prefix);
  uint64 time = clock_.Get();
  File::Write(prefix + ".time", Memblock((const char *)&time, sizeof(uint64)));
  uint64 time1 = TimeNS();
  cout << "Economy::Store:: time (s): " 
       << (time1 - time0) * 1e-9 << endl;
  sleep(1);
}

void Economy::CheckRunFile() {
  if (!FileExists("run")) return;
  {
    ifstream input("run");
    char buf[1000];
    while (input.getline(buf, 1000)) {
      ScheduleCommand(buf);
    }
  }
  // compiler warns on unused result. 
  // the following line may make a little less sense
  if (system("mv run run.last") == 999101010) 
    cout << "lucky lucky" << endl;
}

void * ThresholdLoop(void *junk) {
  while (1) {
    // these are in seconds.
    uint64 explore_time = ConfigInt("explore_time");
    uint64 exploit_time = ConfigInt("exploit_time");
    uint64 best_threshold = 0;
    int64 biggest_improvement = 0;
    for (int64 threshold = 12; threshold >=0; threshold--) {
      G_EC->accept_threshold_ = pow((float)10.0, (float)threshold);
      int64 old_util = G_EC->utility_;
      usleep((explore_time * 1000000) / 13);
      int64 improvement = G_EC->utility_ - old_util;
      if (improvement > biggest_improvement) {
	biggest_improvement = improvement;
	best_threshold = threshold;
      }
      /*
      COUT << "Threshold " << threshold << " improvement " << improvement 
	   << "\n";
      */
    }
    COUT << "Best Threshold: " << best_threshold << "\n";
    G_EC->accept_threshold_ = pow((float)10.0, (float)best_threshold);
    usleep(exploit_time * 1000000);
  }
  return NULL;
}

void * CommandsLoop(void *junk) {
  while (1) {
    sleep(1);
    G_EC->CheckRunFile();
    G_EC->CheckCronFile();
  }
  return NULL; }

bool StartsWith(string s, string t) {
  return (s.size() >= t.size()) && (s.substr(0, t.size()) == t);
}

void Economy::ScheduleCommand(string s) {
  if (StartsWith(s, "nostop")) {
    RunCommand(s);
    return;
  };
  cout << "::ScheduleCommand " << s << endl;
  MutexLocker locker(&commands_lock_);
  last_run_[s] = 0x9999999999999999ull;
  commands_.push_back(s);
  G_SYNC->StopAll();
}

void Economy::RunScheduledCommands()  { 
  vector<string> v;
  {
    MutexLocker locker(&commands_lock_);
    v = commands_;
    commands_.clear();
  }
  for (uint64 i=0; i<v.size(); i++) RunCommand(v[i]);
}

void Economy::RunCommand(string s) {
  istringstream input(s);
  string command;
  input >> command;
  if (command == "nostop") input >> command;
  if (command == "accept-threshold") {
    uint64 t;
    if (!(input >> t)) t = rand() % 13;
    G_EC->accept_threshold_ = pow((float)10.0, (float)t);
  }
  if (command == "store") {
    string prefix;
    input >> prefix;
    Store(prefix);
  }
  if (command == "die") {
    exit(1);
  }
  last_run_[s] = time(0);
}

void Economy::CheckCronFile() {
  static uint64 first_cron_time = time(0);
  uint64 now = time(0);
  
  ifstream input("cron");
  char buf[1000];
  while (input.getline(buf, 1000)) {
    string sbuf = buf;
    istringstream ibuf(sbuf);
    uint64 t;
    if (!(ibuf >> t)) continue;
    if ( t== 0) continue;
    ibuf >> ws;
    string cmd;
    string s;
    while (ibuf >> s) cmd = cmd + " " + s;
    if ((max(last_run_[cmd], first_cron_time) + t) < now) {
      ScheduleCommand(cmd);
    }
  }
}

void EconomyThread::SetThreadAllocator(ThreadAllocator *a) {
  a_ = vi_->a_ = a;
  // we shouldn't need this
  // tester_->SM()->SetThreadAllocator(a);
}

// Pass in NULL as the requested_creation_time to use now
// Most callers should use that, except for REPOST which
// uses the older time
void EconomyThread::PushAd(Thing ad, int64 weight) {
  CHECK(weight >= 0);

  if (!ad.IsArray() || 
      (ad.ArraySize() < 2)) return;
  
  Thing lhs = ad.Ref(0);
  Thing rhs = ad.Ref(1);
  
  // maybe ignore if we already have ads with the given lhs, rhs
  uint64 weight_so_far
    = e_->ad_store_.RangeWeight
    (Ads::Prefix(2, lhs.ID(), rhs.ID()), ad_store_reader_);
  if (weight_so_far) weight = rng_.Gamble(weight, weight_so_far);
  if (weight == 0) return;

  // Do we need to construct a local_ads_
  if (!local_ads_) {
    local_ads_ = new Ads::Sampler;
    local_ads_->Allocate(local_ads_capacity_);
  }

  // Do we need to increase weight for memory allocated
  uint64 global_words_allocated_start = a_->global_words_allocated_;
  ad = a_->Globalize(ad);
  uint64 global_words_allocated 
    = a_->global_words_allocated_ - global_words_allocated_start;
  int64 min_weight = (3 + global_words_allocated) * kMinWeightPerWord;
  if ((weight < min_weight) && (lhs != IMPROVEMENT)) {
    weight += Sleep(min_weight - weight);
    if (weight < min_weight) return;
  }

  // Create and push back the tuple locally
  Ads::Tuple tuple = 
    Ads::FullTuple(weight, lhs.ID(), rhs.ID(), ad.ID());
  local_ads_->AddTuple(tuple);
  
  // Do we need to hand over our sampler
  if (local_ads_->IsFull()) FlushAds(false);
}

void EconomyThread::FlushAds(bool force) {
  if (local_ads_ && local_ads_->size_) {
    local_ads_->Build();
    local_ads_q_.push_back(local_ads_);
    local_ads_ = NULL;
  }
  if (local_ads_q_.size() == 0) return;
  force = force || (local_ads_q_.size() == local_ads_q_.capacity_);
  if (e_->ad_store_.AddSamplers(local_ads_q_.begin(), local_ads_q_.size(), 
				force))
    local_ads_q_.clear();
}

bool EconomyThread::ReadAd(Thing lhs, Thing *rhs) {
  Ads::Tuple begin = Ads::Prefix(1, lhs.ID());
  Ads::Tuple end = Ads::Prefix(1, lhs.ID() + 1);
  Ads::Tuple * tuple = 
    e_->ad_store_.Sample
    (begin, end, ad_store_reader_, &rng_, Ads::SAMPLE_UNIFORM_OVER_SAMPLERS);
  if (tuple == NULL) return false;
  *rhs = tuple->a_[1];
  return true;
}

// Grab a contributor linked list for this (lhs, rhs)
// from a post with creation-time before end_time
// returns (<contributor-linked-list-of-entities> <post-time>)
Thing EconomyThread::ReadContributors
(Thing lhs, Thing rhs, Thing end_time) {
  uint64 end_time_i = ~0ull;
  if (end_time.IsInteger()) end_time_i = end_time.GetInteger();
  Ads::Tuple start = Ads::FullTuple(0, lhs.ID(), rhs.ID(), 0, 0);
  Ads::Tuple end = Ads::FullTuple(0, lhs.ID(), rhs.ID(),
				  Thing::MakeInteger(end_time_i).ID(), 0);
  Ads::Tuple * tuple = 
    e_->ad_store_.Sample
    (start, end, ad_store_reader_, &rng_, Ads::SAMPLE_WEIGHTED);
  if (tuple == NULL) return NULL;
  return tuple->a_[3]; // the contributor list
}

EconomyThread::EconomyThread() 
  : local_ads_(NULL)
  , local_ads_q_(1 << ConfigInt("lg_local_ads_max_samplers")) {
  vi_ = new Vinterpreter;
  vi_->system_ = this;
  e_ = G_EC;
  {
    MutexLocker ml(&e_->lock_);
    e_->et_.insert(this);
  }
  tester_ = new SolutionThread(e_->solution_, this);
  local_ads_capacity_ = 1 << ConfigInt("lg_local_ads_capacity");
  ad_store_reader_ = new RWLock::Reader(e_->ad_store_.sample_lock_);
}

EconomyThread::~EconomyThread() {
  // cout << "~EconomyThread" << endl;
  delete vi_;
  {
    MutexLocker ml(&e_->lock_);
    e_->et_.erase(this); 
  }
  delete tester_;
}

bool EconomyThread::ReadInitialAd(istream& in, Thing *ad, 
				  uint64 *weight) {
  double weight_per_thread =
    ConfigDouble("default_initial_weight_per_thread");
  uint64 default_weight =
    ConfigInt("num_economy_threads") * weight_per_thread;
  in >> ws;
  *weight = default_weight;
  *ad = ReadThing(a_, in);
  if (ad->IsNot()) return false;
  
  // Read flags
  if (*ad == WEIGHT) {
    double d_weight = ReadThing(a_, in).GetDouble();
    *weight = (uint64) (d_weight * default_weight);
    *ad = ReadThing(a_, in);      
  }
  in >> ws;
  CHECK(*weight >= 0);
  return true;
}


void EconomyThread::ReadInitialAds(string filename) {
  ifstream initial_in(filename.c_str());
  Thing ad;
  uint64 weight;

  while (ReadInitialAd(initial_in, &ad, &weight)) {
    PushAd(ad, weight);
  }
  FlushAds(true);
  e_->ad_store_.MakePermanent();
  CHECK(e_->ad_store_.num_permanent_ == 1);
}

// For now has to be a constant payout
int64 EconomyThread::Payout() { return 10000; }

Thing EconomyThread::SystemCall(int32 which, int32 num_parameters) {
  vi_->NoFunctionCache();
  Syscall call = (Syscall) which;
  switch (call) {
  case SYSCALL_POST: {
    vi_->NoFunctionCache();
    if (num_parameters != 2) throw ARITY_EXCEPTION;
    Thing ad = vi_->GetParameter(0, num_parameters);
    Thing cash = vi_->GetParameter(1, num_parameters);
    Post(ad, cash);
    return NIL;
  }
  case SYSCALL_SAMPLE: {
    vi_->NoFunctionCache();
    if (num_parameters != 3) throw ARITY_EXCEPTION;
    Thing inclusive_start = vi_->GetParameter(0, num_parameters);
    Thing inclusive_end = vi_->GetParameter(1, num_parameters);
    Thing strategy_thing = vi_->GetParameter(2, num_parameters);
    return SyscallSample(inclusive_start, inclusive_end, strategy_thing);
  }
  case SYSCALL_RANGEWEIGHT: {
    vi_->NoFunctionCache();
    if (num_parameters != 2) throw ARITY_EXCEPTION;
    Thing inclusive_start = vi_->GetParameter(0, num_parameters);
    Thing inclusive_end = vi_->GetParameter(1, num_parameters);
    return SyscallRangeWeight(inclusive_start, inclusive_end);
  }
  case SYSCALL_RANGECOUNT: {
    vi_->NoFunctionCache();
    if (num_parameters != 2) throw ARITY_EXCEPTION;
    Thing inclusive_start = vi_->GetParameter(0, num_parameters);
    Thing inclusive_end = vi_->GetParameter(1, num_parameters);
    return SyscallRangeCount(inclusive_start, inclusive_end);    
  }
  case SYSCALL_READALL: {
    vi_->NoFunctionCache();
    if (num_parameters != 2) throw ARITY_EXCEPTION;
    Thing inclusive_start = vi_->GetParameter(0, num_parameters);
    Thing inclusive_end = vi_->GetParameter(1, num_parameters);
    return SyscallReadAll(inclusive_start, inclusive_end);
  }
  case SYSCALL_TEST: {
    vi_->NoFunctionCache();
    if (num_parameters < 1) throw ARITY_EXCEPTION;
    Thing contributors = vi_->GetParameter(0, num_parameters);
    if (num_parameters >= 2)
      return Test(contributors, 
		  vi_->GetParameter(1, num_parameters));
    else
      return Test(contributors);
  }
  case SYSCALL_FLUSH:
    vi_->NoFunctionCache();
    FlushAds(true);
    return NIL;
  default:
    Thing ret = tester_->SystemCall(call, num_parameters);
    if (ret != NOTIMPLEMENTED) return ret;
    throw UNKNOWN_SYSCALL_EXCEPTION;
  }
}

// Take time from the interpreter, and tell us how much you took
int64 EconomyThread::Sleep(int64 request) {
  int64 time_remaining = vi_->TimeRemaining();
  if (request > time_remaining) request = time_remaining;
  if (request < 0) return 0;
  vi_->clock_.Advance(request);
  return request;
}

// Verifies parameters, subtracts time from the clock
// then calls PushAd
void EconomyThread::Post(Thing ad, Thing cash) {
  // CHECK(contributors_ != NIL);
  if (!cash.IsInteger()) throw TYPE_EXCEPTION;
  if (ad.NodeCount() == 0) return;
  if (!ad.IsArray() || (ad.ArraySize() < 2)) {
    cout << "bad ad = " << ad << endl;
    throw TYPE_EXCEPTION;
  }
  CHECK(ad.IsArray())
  int64 cash_actual = Sleep(cash.GetInteger());
  if (cash_actual <= 0) return;
  PushAd(ad, cash_actual);
}

/*
Thing EconomyThread::SyscallRepost
(Thing lhs, Thing demand, Thing rhs, Thing end_time, Thing cash) {
  if (!end_time.IsInteger()) {
    COUT << "end_time error " << end_time << "\n";
    throw TYPE_EXCEPTION;
  }
  if (!cash.IsInteger()) {
    COUT << "cash error " << cash << "\n";
    throw TYPE_EXCEPTION;
  }
  int64 cash_i = cash.GetInteger();
  if (cash_i < 0) {
    // cout << "::repost cash_i < 0" << endl;
    return NIL;
  }
  int64 cash_actual = Sleep(cash_i);
  int64 end_time_i = end_time.GetInteger();
  if (end_time_i == 0) throw UNKNOWN_EXCEPTION;
  Ads::Tuple start = Ads::FullTuple(0, lhs.ID(), demand.ID(), rhs.ID(), 0, 0, 0);
  Ads::Tuple end = Ads::FullTuple(0, lhs.ID(), demand.ID(), rhs.ID(),
				  Thing::MakeInteger(end_time_i).ID(), 0, 0);
  Ads::Tuple * tuple = 
    e_->ad_store_.Sample
    (start, end, ad_store_reader_, &rng_, Ads::SAMPLE_WEIGHTED);

  if (!tuple) {
    Thing push_time = Thing::MakeInteger(end_time_i - 1);
    Thing ret = 
      PushAd(lhs, demand, rhs, EmptyContributors(), cash_actual, push_time);
    // if (ret == NIL) cout << "::repost !tuple" << endl;
    return ret;
  }

  Thing ret = 
    PushAd(tuple->a_[0], tuple->a_[1], tuple->a_[2], 
	   tuple->a_[4], cash_actual, tuple->a_[3]);
  // if (ret == NIL) cout << "::repost PushAd" << endl;
  return ret;
}
*/

// Returns the amount of improvement to the utility
Thing EconomyThread::Test(Thing contributors, Thing maybe_force) {
  // cout << "::EconomyThread::Test" << endl;

  Thing diff_thing = tester_->Test(maybe_force);
  if (diff_thing == SYSTEM_NODEBUG) {
    // cout << "::Test preempted" << endl;
    return diff_thing;
  }
  int64 diff = diff_thing.GetInteger();
  if (diff <= 0) {
    // cout << "::Test diff = " << diff << endl;
    return Thing::MakeInteger(0);
  }

  // Figure out credit in global nanoseconds
  double credit = 0.0;
  {
    MutexLocker ml(&e_->credit_lock_);
    e_->utility_ += diff;
    credit = 1000 * 
      e_->assigner_.Credit(e_->clock_.Get(), 
			   (double)e_->utility_);
  }  
  CHECK(credit >= -0.01);

  Thing improvement_ad = 
    a_->MakeImmutableArray3
    (LOCAL_ALLOCATION, IMPROVEMENT, 
     Thing::MakeInteger(G_EC->clock_.Get()), 
     contributors);
  
  PushAd(improvement_ad, (uint64) credit);

  // Print how long the entity has run for till now 
  // uint64 economic_running_time = vi_->clock_.Get();
  uint64 real_running_time = vi_->wall_clock_.Get();
  if ((rng_.Random() % 100) == 0)
  cout << "Credit: " << credit << endl
    // << " Economic Time (ns): " << Delimit(economic_running_time) << endl
       << " Real Time (ns): " << Delimit(real_running_time) << endl;
  FlushAds(true);
  return Thing::MakeInteger(diff);
}

bool ArrayThingToPrefix(Thing a, Ads::Tuple * result) {
  if (!a.IsArray()) return false;
  if (a.ArraySize() >= Ads::WIDTH) return false;
  for (uint64 i=0; i<a.ArraySize(); i++) result->a_[i] = a.Ref(i).ID();
  result->prefix_length__ = a.ArraySize();
  return true;
}

// return an ad in the given range. 
// inclusive_start and inclusive_end are prefixes of ads.
// strategy_thing is an integer corresponding to a SampleStrategy
Thing EconomyThread::SyscallSample(Thing inclusive_start,
				   Thing inclusive_end,
				   Thing strategy_thing){
  Ads::Tuple start_prefix, end_prefix;
  if (!ArrayThingToPrefix(inclusive_start, &start_prefix)) return NULL;
  if (!ArrayThingToPrefix(inclusive_end, &end_prefix)) return NULL;
  if (!strategy_thing.IsInteger()) return NULL;
  uint64 strategy = strategy_thing.GetInteger();
  if (strategy >= Ads::SAMPLE_NUM_STRATEGIES) return NULL;
  Ads::Tuple * tuple 
    = e_->ad_store_.Sample(start_prefix.PrefixToLowerBound(), 
			   end_prefix.PrefixToUpperBound(), 
			   ad_store_reader_, &rng_, 
			   Ads::SampleStrategy(strategy));
  if (tuple == NULL) return NIL;
  return Thing(tuple->a_[Ads::AD]);
}

Thing EconomyThread::SyscallRangeWeight(Thing inclusive_start, 
					Thing inclusive_end) {
  Ads::Tuple start_prefix, end_prefix;
  if (!ArrayThingToPrefix(inclusive_start, &start_prefix)) return NULL;
  if (!ArrayThingToPrefix(inclusive_end, &end_prefix)) return NULL;
  int64 rw = 
    e_->ad_store_.RangeWeight(start_prefix.PrefixToLowerBound(), 
			      end_prefix.PrefixToUpperBound(), 
			      ad_store_reader_);
  return Thing::MakeInteger(rw);
}

Thing EconomyThread::SyscallRangeCount(Thing inclusive_start, 
				       Thing inclusive_end) {
  Ads::Tuple start_prefix, end_prefix;
  if (!ArrayThingToPrefix(inclusive_start, &start_prefix)) return NULL;
  if (!ArrayThingToPrefix(inclusive_end, &end_prefix)) return NULL;
  Ads::Tuple lbound = start_prefix.PrefixToLowerBound();
  Ads::Tuple ubound = end_prefix.PrefixToUpperBound();

  RWLock::ReadLocker locker(ad_store_reader_);
  int64 range_size = e_->ad_store_.RangeSizeLocked(lbound, ubound);
  return Thing::MakeInteger(range_size);
}

Thing EconomyThread::SyscallReadAll(Thing inclusive_start, 
 				    Thing inclusive_end) {
  Ads::Tuple start_prefix, end_prefix;
  if (!ArrayThingToPrefix(inclusive_start, &start_prefix)) return NULL;
  if (!ArrayThingToPrefix(inclusive_end, &end_prefix)) return NULL;
  Ads::Tuple lbound = start_prefix.PrefixToLowerBound();
  Ads::Tuple ubound = end_prefix.PrefixToUpperBound();

  RWLock::ReadLocker locker(ad_store_reader_);
  //  int64 start_num_samplers = 
  //  e_->ad_store_.NumSamplers();
  int64 range_size = e_->ad_store_.RangeSizeLocked(lbound, ubound);
  // TODO: check if we have enough time.
  Thing ret = a_->MakeMutableArray();
  Box * b = ret.GetBox();
  a_->ResizeMutableArray(b, range_size * 2);
  if (range_size == 0) return ret;
  uint64 * p_start = (uint64 *)(b->MutableArrayData());
  uint64 * p = p_start;
  uint64 * end_p = p_start + range_size * 2;
  e_->ad_store_.CopyOutRangeLocked(lbound, ubound, &p, end_p);
  if (p < end_p) a_->ResizeMutableArray(b, p - p_start);
  for (uint64 i=0; i<ret.ArraySize(); i += 2) {
    Thing ad(ret.Ref(i));
    Thing weight(ret.Ref(i+1));
    if (!(weight.IsInteger())) {
      cerr << "inclusive_start " << inclusive_start << endl;
      cerr << "inclusive_end " << inclusive_end << endl;
      cerr << "lbound " << lbound.ToString() << endl;
      cerr << "ubound " << ubound.ToString() << endl;
      cerr << "i = " << i << endl;
      cerr << "range_size = " << range_size << endl;
      for (int j=0; j<2 * range_size; j++) {
	cerr << j << " " << Thing(p[j]) << endl;
      } 
    }
    CHECK(weight.IsInteger());
    CHECK(ad.IsArray());
    CHECK(ad.ArraySize() >= 2);
  }
  return ret;
}


void EconomyThread::Loop(int64 max_cpu) {  
  while (1) {
    try {
      ThreadAllocator ta;
      SetThreadAllocator(&ta);
      while (1) {
	RunLispCode();
	ta.ClearLocalBlock();
      }
    } 
    catch (LocalAllocatorException) {
      cout << "*** Local Allocator Exception ***\n"; }
    catch (StableVectorException) {}
    catch (SyncException) {}
    catch (CollisionException) {
      cout << "*** Collision Exception ***" << endl; } 
    catch (HashTable::FullException) {
      cout << "*** Full Exception ***" << endl; 
      CHECK(false);
    }
  }
}

string EconomyThread::Serve(WebRequest request) {  
  try {
    RWLock::WriteLocker l(&tester_->reader_);
    ThreadAllocator ta;
    ta.ReallocateLocalBlock
      (1ull << 
       (ConfigInt("lg_local_allocation_block_size") +
	ConfigInt("lg_user_local_allocation_block_increment")));
    SetThreadAllocator(&ta);
    ostringstream ostr;
    ostr << "( serve '( ";    
    forall(run, request.params_) 
      ostr << "( " << run->first << " " << run->second << " )";
    ostr << ") )";
    // cout << "parsing" << ostr.str() << endl;
    Thing code = ToThing(&ta, ostr.str());
    if (code.IsNot()) return "unparsable " + ostr.str();
    Thing creation_time = Thing::MakeInteger(G_EC->clock_.Get());
    Thing ret = RunLispCode(code);
    if (ret.IsNot()) return "Returned NULL";
    return (ret.IsString()) ? 
      StringToCPPString(ret) : ("not a string " + ToString(ret));
  } 
  catch (LocalAllocatorException) {
    cout << "*** LocalAllocatorException ***" << endl;
  }
  catch (StableVectorException) {}
  catch (SyncException) {
  } catch (CollisionException) {
    cout << "*** Collision Exception ***" << endl;
  } catch (HashTable::FullException) {
    cout << "*** Full Exception ***" << endl;
    CHECK(false);
  }
  return "Caught Exception";
}

// Print immutable information about an entity
string EconomyThread::EntityToString(Thing & e) {  
  return "TODO"; }

// TODO: shift some of the resposibility for limiting to the lisp code. 
int64 EconomyThread::RandomTimeLimitNS() {
  // 10 ms -> 10 s
  return uint64 (1e7 / max(rng_.RandomFraction(), 1e-3));
}

// Run an actor and try and do something
// Pass in code or pass in null to run a random rhs from 'code
Thing EconomyThread::RunLispCode(Thing code) {
  // Figure out what's running and it's time limit
  int64 time_limit = (int64) 300e9;
  if (code.IsNot()) time_limit = RandomTimeLimitNS();

  running_ = code;
  while (running_.IsNot()) ReadAd(CODE, &running_); 
  
  // Run it until it runs out of time
  Thing return_value = NULL;
  vi_->wall_clock_.Set(0);
  vi_->clock_.Set(0);
  tester_->NewOverlayCall(NIL);
  return_value = 
    vi_->SafeExecute(running_, time_limit, kEconomyMode, NIL);
  a_->ClearLocalBlock();
  return return_value;
}

int credit_main() {
  CreditAssigner ca;
  double t, u;  
  while (cin >> t >> u) {
    cout << ca.Credit(t, u) << endl;
  }
  return 0;
}

void * MakeLoopEconomyThread(void * junk) {
  EconomyThread et; 
  et.Loop(-1);
  CHECK(false); return NULL;
}

void Economy::StartWebServerThread() { 
  EconomyThread * t = new EconomyThread;
  WebServer * server = new WebServer(ConfigInt("server_port"), t);
  server->Launch();
}

void Economy::StartThreads() {
  num_threads_ = ConfigInt("num_economy_threads");
  for (uint64 i=0; i<num_threads_; i++) {
    pthread_t *thread = new pthread_t;
    int ret = pthread_create(thread, NULL, MakeLoopEconomyThread, NULL);
    if (ret) cout << "Error " << ret << endl;
  }
  {
    pthread_t *thread = new pthread_t;
    int ret = pthread_create(thread, NULL, CommandsLoop, NULL);
    if (ret) cout << "Error " << ret << endl;
  }
  if (ConfigString("use_threshold_loop") == "true")
  {
    pthread_t *thread = new pthread_t;
    int ret = pthread_create(thread, NULL, ThresholdLoop, NULL);
    if (ret) cout << "Error " << ret << endl;
  }
  StartWebServerThread();
  // cout << "Started threads" << endl;
}

int loop_economy_main() {
  InitEconomy();
  G_EC->StartThreads();
  sleep(1000000000);
  pthread_exit(NULL); 
  return 0;
}

int output_economy_main() { return 0; }

int interpret_economy_main() {
  InitEconomy();
  G_EC->StartWebServerThread();
  EconomyThread et;
  ThreadAllocator ta;
  ta.ReallocateLocalBlock
    (1ull << 
     (ConfigInt("lg_local_allocation_block_size") +
      ConfigInt("lg_user_local_allocation_block_increment")));
  et.SetThreadAllocator(&ta);
  Thing code;
  while (1) {
    cin >> ws;
    if (cin.peek() == '~') {
      cin.get();
      char buf[1000];
      cin.getline(buf, 1000);
      G_EC->RunCommand(string(buf));
      continue;
    }
    code = ReadThing(et.a_, cin);
    if (code.IsNot()) break;
    Thing creation_time = Thing::MakeInteger(G_EC->clock_.Get());
    uint64 start = TimeNS();
    Thing ret = et.RunLispCode(code);    
    cout << ToString(ret) << endl;
    cout << "Wall Time(ns) = " << Delimit(TimeNS() - start) << endl;
  }
  sleep(1000000);
  return 0;
}

int serve_economy_main() {
  InitEconomy();
  G_EC->StartWebServerThread();
  sleep(1000000);
  return 0;
}

int prepare_main() {
  InitJobQueue();
  PrepareDataSet(ConfigString("corpus_file"));
  return 0;
}

int time_test_main() {
  InitJobQueue();
  TimeTest();
  return 0;
}

int main(int argc, char ** argv) {
  InitUtil(argc, argv);
  int64 sr = ConfigInt("srand");
  if (!sr) sr = time(0);
  srand(sr);
  cout << "sr: " << sr << endl; 
  string command;
  if (argc > 1) command = argv[1];
  if (command == "in") return interpret_main();
  if (command == "credit") return credit_main();
  if (command == "loop") return loop_economy_main();
  if (command == "prepare") return prepare_main();
  if (command == "prepare_integers") { 
    IntegerMakeExplicitProblemFile(); return 0; }
  if (command == "make_binary_sequence_file")
    { MakeBinarySequenceFile(); return 0; }
  if (command == "output") return output_economy_main();
  if (command == "ie") return interpret_economy_main();
  if (command == "tok") return tokenize_main();
  if (command == "lex") return lexicon_main();
  if (command == "sort") return integerarray_main();
  if (command == "file") return file_main();
  // if (command == "rw") return test_rw_lock_main();
  // if (command == "set") { TestSetModel(); return 0; } 
  if (command == "serve") { serve_economy_main();  return 0; }
  if (command == "vl") { vl_encode_main(); return 0; } 
  if (command == "math") {
    return math_main();
  }
  if (command == "tableset") {
    return 0;
    // return TableSetTest();
  }
  if (command == "time") { return time_test_main(); } 
  return loop_economy_main();
}

void Economy::AboutToCollect() {
  forall(run, et_) (*run)->FlushAds(true);
  ad_store_.Shrink();
}

void Economy::FlagReferences(int32 shard, Frontier * frontier) {
  // Flag references for the ads system
  for (uint64 i=0; i<ad_store_.samplers_.size(); i++) {
    Ads::Sampler *s = ad_store_.samplers_[i];    
    for (uint64 j=ShardBegin(s->size_, NumShards(), shard); 
	 j<ShardEnd(s->size_, NumShards(), shard); j++) {
      Ads::Tuple & t = s->data_[j];
      for (uint64 k=0; k<Ads::WIDTH; k++) frontier->Add(Thing(t.a_[k]));
    }
  }
}


